package com.da.javatest.thread.juc.countdownlatch;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.springframework.util.CollectionUtils;

/**
 * @author chenlida
 * @date 2020/7/12 12:00
 * 并行处理任务的工具类
 */
public class TaskDisposeUtils {
    /**
     * 并行线程数
     */
    private static final int POOL_SIZE;
    private static final int MAX_POOL_SIZE;

    static {
        POOL_SIZE = Integer.max(Runtime.getRuntime().availableProcessors(), 5);
        MAX_POOL_SIZE = Integer.max(POOL_SIZE, 10);
    }

    private static ExecutorService executorService = new ThreadPoolExecutor(POOL_SIZE, MAX_POOL_SIZE, 0L,
        TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new BasicThreadFactory.Builder()
        .namingPattern("taskDisposeUtil-thread-pool-%d").daemon(true).build());

    /**
     * 并行处理，并等待结束
     *
     * @param taskList 任务列表
     * @param consumer 消费者
     * @param <T>
     * @throws InterruptedException
     */
    public static <T> void dispose(List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        dispose(true, taskList, consumer);
    }

    /**
     * 并行处理，并等待结束
     *
     * @param moreThread 是否多线程执行
     * @param taskList   任务列表
     * @param consumer   消费者
     * @param <T>
     * @throws InterruptedException
     */
    private static <T> void dispose(boolean moreThread, List<T> taskList, Consumer<T> consumer)
        throws InterruptedException {
        if (CollectionUtils.isEmpty(taskList)) {
            return;
        }
        if (moreThread) {
            CountDownLatch countDownLatch = new CountDownLatch(taskList.size());
            for (T item : taskList) {
                executorService.execute(() -> {
                    try {
                        consumer.accept(item);
                    } finally {
                        countDownLatch.countDown();
                    }
                });
            }
            countDownLatch.await();
        } else {
            for (T item : taskList) {
                consumer.accept(item);
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        //生成1-10的10个数字，放在list中，相当于10个任务
        List<Integer> list = Stream.iterate(1, a -> a + 1).limit(10).collect(Collectors.toList());
        //启动多线程处理list中的数据，每个任务休眠时间为list中的数值
        TaskDisposeUtils.dispose(list, item -> {
            try {
                long startTime = System.currentTimeMillis();
                TimeUnit.SECONDS.sleep(item);
                long endTime = System.currentTimeMillis();

                System.out.println("任务" + item + "执行完成，耗时：" + (endTime - startTime));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        //上面所有任务处理完毕完毕之后，程序才能继续
        System.out.println(list + "中的所有任务处理完成");
    }
}
